package com.rs.controller;

import com.alibaba.fastjson.JSONObject;
import com.rs.mapper.CommonSearchMapper;
import dm.jdbc.driver.DmdbBlob;
import dm.jdbc.driver.DmdbNClob;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;

@RestController
public class KafkaTestController {

    // 1、声明topic
    private static final String TOPIC_NAME = "test_20230213";

    // 2、注入kafka
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    // 3、模拟发送消息
    @PostMapping("/api/v1/sendMessage")
    public void sendMessage(@RequestBody JSONObject json) {
        kafkaTemplate.send(TOPIC_NAME, json.toJSONString()).addCallback(success -> {
            assert success != null;
            // 消息发送的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功" + topic + "-" + partition + "-" + offset);
        }, failure -> System.out.println("发送消息失败：" + failure.getMessage()));
    }

    @Resource
    private CommonSearchMapper commonSearchMapper;

    @PostMapping("/api/v1/search")
    public List<Map<String, Object>> search(String tableName) throws SQLException {
        String sql = "select * from " + tableName;

        List<Map<String, Object>> list = commonSearchMapper.divSql(sql);

        for (Map<String, Object> map : list) {
            for (Map.Entry<String, Object> entry : map.entrySet()) {
                Object value = entry.getValue();
                if (value instanceof DmdbBlob) {
                    String myBlob = new String(((DmdbBlob) value).data, StandardCharsets.UTF_8);
                    map.put(entry.getKey(), myBlob);
                } else if (value instanceof DmdbNClob) {
                    map.put(entry.getKey(), ((DmdbNClob) value).data);
                }
            }
        }

        return list;
    }
}
